/*
 * Copyright (c) 2012-2015 The original author or authors
 * ------------------------------------------------------
 * All rights reserved. This program and the accompanying materials
 * are made available under the terms of the Eclipse Public License v1.0
 * and Apache License v2.0 which accompanies this distribution.
 *
 * The Eclipse Public License is available at
 * http://www.eclipse.org/legal/epl-v10.html
 *
 * The Apache License v2.0 is available at
 * http://www.opensource.org/licenses/apache2.0.php
 *
 * You may elect to redistribute this code under either of these licenses.
 */
package com.chinamobile.iot.mqtt.spi.impl;

import com.alibaba.fastjson.JSON;
import com.chinamobile.iot.command.CommandCache;
import com.chinamobile.iot.config.MqttConfiguration;
import com.chinamobile.iot.handler.model.SubMessageData;
import com.chinamobile.iot.mqtt.interception.MessageHandler;
import com.chinamobile.iot.mqtt.schedule.DelayTask;
import com.chinamobile.iot.mqtt.server.netty.NettyUtils;
import com.chinamobile.iot.mqtt.spi.model.Qos1Message;
import com.chinamobile.iot.mqtt.spi.model.StoredMessage;
import com.chinamobile.iot.mqtt.spi.model.Tree1Client;
import com.chinamobile.iot.mqtt.spi.model.WaitAckMessage;
import com.chinamobile.iot.mqtt.spi.security.IAuthenticator;
import com.chinamobile.iot.mqtt.spi.security.IAuthorizator;
import com.chinamobile.iot.mqtt.spi.security.MqttClientInfo;
import com.chinamobile.iot.utils.NameThreadFactory;
import io.netty.buffer.ByteBuf;
import io.netty.buffer.Unpooled;
import io.netty.channel.Channel;
import io.netty.channel.ChannelPipeline;
import io.netty.handler.codec.mqtt.*;
import io.netty.handler.codec.mqtt.MqttConnectReturnCode;
import io.netty.handler.timeout.IdleStateHandler;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.data.redis.core.ListOperations;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.stereotype.Component;
import org.springframework.util.StopWatch;

import javax.annotation.PostConstruct;
import javax.annotation.PreDestroy;
import java.nio.charset.Charset;
import java.util.LinkedList;
import java.util.List;
import java.util.Random;
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.atomic.AtomicInteger;

import static com.chinamobile.iot.handler.redis.RedisConst.REDIS_4_KEY_PREFIX;
import static com.chinamobile.iot.handler.sub.SendSubConst.CLIENT_ID_SEND_SUB;
import static com.chinamobile.iot.handler.sub.SendSubConst.TOPIC_SEND_SUB;
import static io.netty.handler.codec.mqtt.MqttConnectReturnCode.*;
import static io.netty.handler.codec.mqtt.MqttQoS.AT_LEAST_ONCE;
import static io.netty.handler.codec.mqtt.MqttQoS.AT_MOST_ONCE;

/**
 * edit by szl
 * mqtt 协议接入处理逻辑
 */
@Component
public class ProtocolProcessor implements DelayTask.TimerCallback<Qos1Message> {

    private static final Logger logger = LoggerFactory.getLogger(ProtocolProcessor.class);

    /**
     * ClientId --> channel,
     */
    private final ConcurrentHashMap<String, ConnectionDescriptor> clientIdMap = new ConcurrentHashMap<>(10240 * 4, 0.95f, 32);
    /**
     * mqttId --> clientId
     */
    private final ConcurrentHashMap<String, String> mqttIdMap = new ConcurrentHashMap<>(10240 * 4, 0.95f, 32);
    /**
     * clientId --> messageId, the messageId used to send Qos1 message to the client.
     */
    private final ConcurrentHashMap<String, AtomicInteger> msgIdMap = new ConcurrentHashMap<>();

    /**
     * the current client count
     */
    private AtomicInteger clientCount = new AtomicInteger(0);


    /**
     * 记录订阅信息的订阅树， 只存储本进程下各个客户端的订阅, client部分为 0ClientId1 表示clientId1以cleanSession 订阅了某个topic
     * ，1ClientId1表示ClientId1以CleanSession为true订阅了某个topic
     */
    private SubscriptionTree<Tree1Client> tree1;

    /**
     * 用于执行一些后台任务
     */
    private ExecutorService executorService;
    /**
     * login authenticator
     */
    @Autowired
    private IAuthenticator authenticator;
    /**
     * publish or subscribe topic authorizator
     */
    @Autowired
    private IAuthorizator authorizator;


    /**
     * the subMessageSaveTask to check if some client ack the QoS1 message
     */
    private DelayTask<WaitAckMessage> subMessageSaveTask;


    @Autowired
    private CommandCache commandCache;

    @Autowired
    private MqttConfiguration mqttConfig;

    /**
     * mqtt 协议消息处理器，包括CLeanSession，订阅保存，分布式相关的等等
     */
    @Autowired(required = false)
    @Qualifier("extendMessageHandler")
    private MessageHandler messageHandler;

    /**
     * mqtt协议的数据处理器，包括数据存储到kafka等等
     */
    @Autowired(required = false)
    @Qualifier("dataHandler")
    private MessageHandler dataHandler;


    @Autowired
    private StringRedisTemplate redisTemplate;


    public ProtocolProcessor() {

    }


    @Override
    public void onTimer(long time, Qos1Message qos1Message, DelayTask<Qos1Message> task) {

    }

    @PostConstruct
    public void init() {
        logger.info("ProtocolProcessor::init() is called!");
        if (mqttConfig.isSubscribe()) {
            this.tree1 = new SubscriptionTree<>();
        }

        executorService = Executors.newFixedThreadPool(10, new NameThreadFactory("protocol-back-"));

        if (mqttConfig.isCleanSession()) {
            subMessageSaveTask = new DelayTask<>(new SubMessageSaveCallback(this, redisTemplate
                    , true), 20, 1, "sub-msg-save");
        } else {
            subMessageSaveTask = new DelayTask<>(new SubMessageSaveCallback(this, redisTemplate
                    , false), 20, 1, "sub-msg-save");
        }
        subMessageSaveTask.start();
    }

    public Channel getChannel(String clientId) {
        ConnectionDescriptor conn = clientIdMap.get(clientId);
        if (conn != null) {
            return conn.channel;
        }
        return null;
    }

    @PreDestroy
    public void destroy() {
        logger.info("ProtocolProcessor::destroy() is called");
        if (mqttConfig.isCleanSession()) {
            subMessageSaveTask.stop();
        }
        executorService.shutdown();
    }

    public SubscriptionTree<Tree1Client> getTree1() {
        return tree1;
    }

    public ConcurrentHashMap<String, ConnectionDescriptor> getClientIdMap() {
        return clientIdMap;
    }

    private StoredMessage asStoredMessage(MqttPublishMessage msg) {
        ByteBuf payload = msg.payload();
        byte[] content = NettyUtils.readBytesAndRewind(payload);
        StoredMessage stored = new StoredMessage(content, msg.fixedHeader().qosLevel(), msg.variableHeader().topicName());
        stored.setRetained(msg.fixedHeader().isRetain());
        if (msg.fixedHeader().qosLevel() != AT_MOST_ONCE) {
            stored.setMessageId(msg.variableHeader().messageId());
        }
        return stored;
    }

    public boolean isOnline(String mqttId) {
        if (mqttId == null) {
            return false;
        }

        if (mqttIdMap.containsKey(mqttId)) {
            return true;
        }

        //todo： 判别代理
        return false;

    }

    public int getClientCount() {
        return clientCount.get();
    }

    public ConcurrentHashMap<String, String> getMqttIdMap() {
        return mqttIdMap;
    }

    public void processConnect(Channel channel, MqttConnectMessage msg) {
        String clientId = msg.payload().clientIdentifier();
        String userName = msg.payload().userName();
        String passWord = msg.payload().password();

        logger.info("processConnect is called, clientId = {}", clientId);

        if (clientId == null) {
            logger.warn("the client is null. channel: {}, it will be disconnected", channel.toString());
            channel.close();
        }

        int keepAlive = msg.variableHeader().keepAliveTimeSeconds();
        boolean cleanSession = msg.variableHeader().isCleanSession();
        // 如果是推送连接登录，直接允许
        if (clientId.equals(CLIENT_ID_SEND_SUB)) {
            executorService.execute(() -> {
                logger.info("send sub client is connected. {}", channel.toString());
                MqttConnAckMessage ok = connAck(CONNECTION_ACCEPTED);
                channel.writeAndFlush(ok);
                // 重置超时检查的时间为keepAlive + 10 秒
                setIdleTime(channel.pipeline(), Math.round(keepAlive + 10));
            });
            return;
        }

        if (this.clientCount.get() >= mqttConfig.getMaxClient()) {
            logger.warn("the connection count reach the max count!");
            channel.close();
            return;
        }

        logger.info("Processing CONNECT message. clientId={}, username={}", clientId, userName);
        if (msg.variableHeader().version() != MqttVersion.MQTT_3_1.protocolLevel()
                && msg.variableHeader().version() != MqttVersion.MQTT_3_1_1.protocolLevel()) {
            executorService.execute(() -> {
                MqttConnAckMessage badProto = connAck(CONNECTION_REFUSED_UNACCEPTABLE_PROTOCOL_VERSION);
                logger.error("MQTT protocol version is not valid. clientId={}", clientId);
                channel.writeAndFlush(badProto);
                channel.close();
            });
            return;
        }


        executorService.execute(() -> {
            StopWatch watch = new StopWatch();
            watch.start();
            String mqttId;
            String rspString;
            MqttClientInfo mqttClientInfo = authenticator.loginCheck(clientId, userName, passWord);
            if (mqttClientInfo == null) {
                MqttConnAckMessage badUser = connAck(CONNECTION_REFUSED_BAD_USER_NAME_OR_PASSWORD);
                logger.warn("clientId: {} login check failed.", clientId);
                channel.writeAndFlush(badUser);
                return;
            }
            watch.stop();
            logger.info("processConnect - auth, clientId = {}, use {}ms", clientId, watch.getLastTaskTimeMillis());

            watch.start();
            mqttId = mqttClientInfo.getId();
            rspString = JSON.toJSONString(mqttClientInfo);
            if (messageHandler != null) {
                // check if there is something wrong with the clientId and the mqttId, for example insert the client to the zookeeper failed.
                if (!messageHandler.onConnecting(clientId, mqttId)) {
                    MqttConnAckMessage badUser = connAck(CONNECTION_REFUSED_BAD_USER_NAME_OR_PASSWORD);
                    logger.warn("clientId: {} login failed. maybe the client is online at other broker", clientId);
                    channel.writeAndFlush(badUser);
                    return;
                }
            }
            watch.stop();
            logger.info("processConnect - onConnecting, clientId = {}, use {}ms", clientId, watch.getLastTaskTimeMillis());

            watch.start();
            ConnectionDescriptor connDescr = new ConnectionDescriptor(channel);
            NettyUtils.clientId(channel, clientId);
            NettyUtils.mqttId(channel, mqttId);
            NettyUtils.cleanSession(channel, cleanSession);

            // reset the timeout second, the value is the keepAlive + 20 seconds
            setIdleTime(channel.pipeline(), (keepAlive * 2 + 10));
            watch.stop();
            logger.info("processConnect - setIdleTime, clientId = {}, use {}ms", clientId, watch.getLastTaskTimeMillis());

            watch.start();
            // check if the client is duplicated login
            if (mqttIdMap.containsKey(mqttId)) {
                logger.info("find another client with the same clientId, the broker will disconnect the old client");
                String oldClientId = mqttIdMap.get(mqttId);
                if (oldClientId != null) {
                    ConnectionDescriptor conn = clientIdMap.get(oldClientId);
                    if (conn != null) {
                        disconnect(conn.channel);
                    }
                }
            }

            ConnectionDescriptor previousConn = clientIdMap.putIfAbsent(clientId, connDescr);
            while (previousConn != null && !previousConn.channel.equals(connDescr.channel)) {
                logger.info("previous Client exist in mqttIdMap.");
                disconnect(previousConn.channel);
                previousConn = clientIdMap.putIfAbsent(clientId, connDescr);
            }

            String previousClient = mqttIdMap.putIfAbsent(mqttId, clientId);
            if (previousClient != null) {
                logger.info("previous mqttId exist in mqttIdMap.");
                ConnectionDescriptor conn = clientIdMap.get(previousClient);
                if (conn != null) {
                    disconnect(conn.channel);
                }
                disconnect(channel);
            }

            // if there are message Id map，use the value as the next msgId of the message
            if (msgIdMap.containsKey(clientId)) {
                AtomicInteger v = msgIdMap.get(clientId);
                connDescr.maxMsgId.set(v.intValue() + 1);
                msgIdMap.remove(clientId);
            }
            watch.stop();
            logger.info("processConnect - recordClient, clientId = {}, use {}ms", clientId, watch.getLastTaskTimeMillis());

            watch.start();
            MqttConnAckMessage ok = connAck(CONNECTION_ACCEPTED);
            channel.writeAndFlush(ok);
            watch.stop();
            clientCount.incrementAndGet();
            logger.info("processConnect - send Connack and recordCount, clientId = {}, use {}ms", clientId, watch.getLastTaskTimeMillis());


            watch.start();
            // notify the observer that a new client is login.
            if (messageHandler != null) {
                messageHandler.onConnected(clientId, mqttId, cleanSession);
            }
            watch.stop();
            logger.info("processConnect - messageHandler.onConnected, clientId = {}, use {}ms", clientId, watch.getLastTaskTimeMillis());
            watch.start();
            if (dataHandler != null) {
                dataHandler.onConnected(clientId, mqttId, cleanSession);
            }
            watch.stop();
            logger.info("processConnect - dataHandler.onConnected, clientId = {}, use {}ms", clientId, watch.getLastTaskTimeMillis());

            watch.start();
            // if needed, then send welcome message to the client
            if (mqttConfig.isSendWelcome()) {
                String topic = authenticator.getRspTopic(mqttClientInfo, clientId);
                // send response message to the client
                if (topic != null) {
                    MqttFixedHeader fixedHeader = new MqttFixedHeader(MqttMessageType.PUBLISH, false, AT_MOST_ONCE, false, 0);
                    MqttPublishVariableHeader varHeader = new MqttPublishVariableHeader(topic, 0);
                    ByteBuf byteBuf = Unpooled.wrappedBuffer(rspString.getBytes());
                    MqttPublishMessage publishMessage = new MqttPublishMessage(fixedHeader, varHeader, byteBuf);
                    channel.writeAndFlush(publishMessage);

                }
            }

            // if this broker enable cleanSession flag and the subscribe, then send the offline message to the client.
            if (mqttConfig.isCleanSession() && mqttConfig.isSubscribe()) {
                ListOperations<String, String> listOperations = redisTemplate.opsForList();
                String dataKey = REDIS_4_KEY_PREFIX + clientId;
                String offlineData = listOperations.leftPop(dataKey);
                int offlineCount = 0;
                while (offlineData != null) {
                    Qos1Message qos1Message = JSON.parseObject(offlineData, Qos1Message.class);
                    StoredMessage storedMessage = qos1Message.getData();
                    int count = route2ClientId(clientId, storedMessage, qos1Message.getRound());
                    // 如果发送不成功，把数据重新放入redis
                    if (count == 0) {
                        listOperations.leftPush(dataKey, offlineData);
                        break;
                    } else {
                        offlineData = listOperations.leftPop(dataKey);
                    }
                    offlineCount++;
                }
            }
            watch.stop();
            logger.info("processConnect - send welcome and offline message, clientId = {}, use {}ms", clientId, watch.getLastTaskTimeMillis());
        });
    }

    private MqttConnAckMessage connAck(MqttConnectReturnCode returnCode) {
        return connAck(returnCode, false);
    }


    private MqttConnAckMessage connAck(MqttConnectReturnCode returnCode, boolean sessionPresent) {
        MqttFixedHeader mqttFixedHeader = new MqttFixedHeader(MqttMessageType.CONNACK, false, AT_MOST_ONCE,
                false, 0);
        MqttConnAckVariableHeader mqttConnAckVariableHeader = new MqttConnAckVariableHeader(returnCode, sessionPresent);
        return new MqttConnAckMessage(mqttFixedHeader, mqttConnAckVariableHeader);
    }

    /**
     * disconnect all theclient connect to this broker
     */
    public void disconnectAllClient() {
        clientIdMap.forEach((k, v) -> v.channel.close());
        clientIdMap.clear();
        mqttIdMap.clear();
        if (tree1 != null){
            tree1.clear();
        }

    }

    public void processPubAck(Channel channel, MqttPubAckMessage msg) {
        String clientID = NettyUtils.clientId(channel);
        String mqttId = NettyUtils.mqttId(channel);
        Boolean cleanSession = NettyUtils.cleanSession(channel);
        int messageId = msg.variableHeader().messageId();
        ConnectionDescriptor connDescr = clientIdMap.get(clientID);
        if (connDescr == null) {
            return;
        }
        if (mqttConfig.isCleanSession() && !cleanSession) {
            WaitAckMessage waitAckMessage = new WaitAckMessage();
            waitAckMessage.setCleanSession(cleanSession);
            Qos1Message qos1Message = new Qos1Message();
            qos1Message.setToClientId(clientID);
            qos1Message.setMsgId(messageId);
            waitAckMessage.setMessage(qos1Message);
            subMessageSaveTask.remove(waitAckMessage);
        }
        executorService.execute(() -> {
            if (messageHandler != null) {
                messageHandler.onMessageAcknowledged(clientID, mqttId, messageId);
            }
            if (dataHandler != null) {
                dataHandler.onMessageAcknowledged(clientID, mqttId, messageId);
            }
        });
    }

    public void processPingReq(Channel channel, MqttMessage msg) {
        executorService.execute(() -> {
            MqttFixedHeader pingHeader = new MqttFixedHeader(
                    MqttMessageType.PINGRESP,
                    false,
                    AT_MOST_ONCE,
                    false,
                    0);
            MqttMessage pingResp = new MqttMessage(pingHeader);
            channel.writeAndFlush(pingResp);
        });
    }

    public void processPublish(Channel channel, MqttPublishMessage msg) {

        StopWatch watch = new StopWatch();
        watch.start();

        final String topic = msg.variableHeader().topicName();
        final Integer messageID = msg.variableHeader().messageId();
        final String content = msg.payload().toString(Charset.forName("UTF-8"));
        byte[] payload = NettyUtils.readBytesAndRewind(msg.payload());


        // check if this message if from other brokers.
        if (topic.equals(TOPIC_SEND_SUB)) {
            logger.info("receive data from other broker. channel: {}", channel.toString());
            // 回复Ack
            executorService.execute(() -> {
                MqttFixedHeader mqttFixedHeader = new MqttFixedHeader(MqttMessageType.PUBACK, false, AT_MOST_ONCE,
                        false, 0);
                MqttMessageIdVariableHeader variableHeader = MqttMessageIdVariableHeader.from(messageID);
                MqttPubAckMessage pubAckMessage = new MqttPubAckMessage(mqttFixedHeader, variableHeader);
                channel.writeAndFlush(pubAckMessage);

                SubMessageData data = JSON.parseObject(payload, SubMessageData.class);
                StoredMessage toStoreMsg = data.getMsg();
                int count = route2Subscribers(toStoreMsg);
                String filter = data.getFilter();
                if (messageHandler != null) {
                    messageHandler.onTopicPublished(null, null, toStoreMsg, filter, count);
                }
            });
            return;
        }
        String clientId = NettyUtils.clientId(channel);
        String mqttId = NettyUtils.mqttId(channel);
        //logger.debug("receive publish message, content = {}, topic = {}, clientId = {}, msgId = {}", content, topic, clientId, messageID);

        MqttQoS qos = msg.fixedHeader().qosLevel();
        if (qos == MqttQoS.EXACTLY_ONCE) {
            logger.info("clientId:{} will be disconnected, the mqtt server dons't support QoS 2, channel:{}", clientId, channel);
            disconnect(channel);
            return;
        }

        if (!authorizator.canWrite(topic, clientId, mqttId, content)) {
            logger.info("clientId:{} will be disconnected, it can't write to topic {}, channel:{}", clientId, topic, channel);
            disconnect(channel);
            return;
        }

        StoredMessage toStoreMsg = asStoredMessage(msg);
        executorService.execute(() -> {
            route2Subscribers(toStoreMsg);
            if (messageHandler != null) {
                messageHandler.onTopicPublished(clientId, mqttId, toStoreMsg, null, 1);
            }
            if (dataHandler != null) {
                dataHandler.onTopicPublished(clientId, mqttId, toStoreMsg, null, 1);
            }
        });

        watch.stop();
        logger.debug("process publish completed. clientId = {}, messageId = {}, use {} ms", clientId, messageID, watch.getLastTaskTimeMillis());
    }


    private void setIdleTime(ChannelPipeline pipeline, int idleTime) {
        if (pipeline.names().contains("idleStateHandler")) {
            pipeline.remove("idleStateHandler");
        }
        pipeline.addFirst("idleStateHandler", new IdleStateHandler(0, 0, idleTime));
    }


    public Integer sendCommandQos1(String clientId, String mqttId, String commandId, String topic, byte[] payload) {
        if (clientId == null) {
            return null;
        }
        ConnectionDescriptor connDescr = clientIdMap.get(clientId);
        if (connDescr == null) {
            logger.warn("send command failed, the clientId:{} is offline", clientId);
            return null;
        }

        if (topic == null || payload == null) {
            logger.error("send command failed, the topic or the payload is null");
            return null;
        }

        Channel channel = connDescr.channel;

        MqttFixedHeader fixedHeader = new MqttFixedHeader(MqttMessageType.PUBLISH, false, AT_LEAST_ONCE, false, 0);
        MqttPublishVariableHeader varHeader = new MqttPublishVariableHeader(topic, connDescr.nextPacketID());
        ByteBuf byteBuf = Unpooled.wrappedBuffer(payload);
        MqttPublishMessage publishMessage = new MqttPublishMessage(fixedHeader, varHeader, byteBuf);

        commandCache.addCommand(clientId, mqttId, varHeader.messageId(), commandId);

        if (channel.isWritable()) {
            logger.info("send a command to client {} success, topic:{}", clientId, topic);
            channel.writeAndFlush(publishMessage);
            return varHeader.messageId();
        } else {
            commandCache.removeCommand(clientId, mqttId, varHeader.messageId());
            logger.info("send a command to client {} failed, topic:{}", clientId, topic);
            return null;
        }
    }

    public String getClientId(String mqttId) {
        String result = null;
        if (mqttId != null) {
            result = mqttIdMap.get(mqttId);
        }
        return result;
    }

    public void notifyChannelWritable(Channel channel) {
//        String clientID = NettyUtils.clientId(channel);
//        ClientSession clientSession = m_sessionsStore.sessionForClient(clientID);
//        boolean emptyQueue = false;
//        while (channel.isWritable() && !emptyQueue) {
//            StoredMessage msg = clientSession.queue().poll();
//            if (msg == null) {
//                emptyQueue = true;
//            } else {
//                // recreate a publish from stored publish in queue
//                MqttPublishMessage pubMsg = createPublishForQos( msg.getTopic(), msg.getQ(), msg.getPayload(),
//                        msg.isRetained(), 0);
//                channel.write(pubMsg);
//            }
//        }
        channel.flush();
    }

    /**
     * when the client is connected to the broker, the broker use this method to send the stored message to the client
     *
     * @param toClientId    the clientid who should receive this message
     * @param storedMessage the stored message
     * @param round         the round of this message to send to the client
     * @return if success, return 1, otherwise return 0
     */
    public int route2ClientId(String toClientId, StoredMessage storedMessage, int round) {
        ConnectionDescriptor conn = clientIdMap.get(toClientId);
        MqttQoS qos = storedMessage.getQos();
        String topic = storedMessage.getTopic();
        ByteBuf byteBuf = storedMessage.getMessage();
        MqttFixedHeader fixedHeader = new MqttFixedHeader(MqttMessageType.PUBLISH, true, qos, false, 0);
        MqttPublishVariableHeader varHeader = new MqttPublishVariableHeader(topic, storedMessage.getMessageId());
        MqttPublishMessage publishMessage = new MqttPublishMessage(fixedHeader, varHeader, byteBuf);
        if (conn != null && conn.channel != null) {
            conn.channel.writeAndFlush(publishMessage);
            boolean cleanSesson = NettyUtils.cleanSession(conn.channel);
            if (mqttConfig.isCleanSession() && qos == AT_LEAST_ONCE && !cleanSesson && round < 3) {
                // add the message to queue to wait for pub ack
                addToMessageSaveTask(toClientId, publishMessage.variableHeader().messageId(), false, storedMessage, round + 1);
            }
            return 1;
        }
        return 0;
    }

    /**
     * Flood the subscribers with the message to notify. MessageID is optional and should only used for QoS 1 and 2
     */
    private int route2Subscribers(StoredMessage pubMsg) {
        if (!mqttConfig.isSubscribe()) {
            return 0;
        }

        final String topic = pubMsg.getTopic();
        final MqttQoS publishingQos = pubMsg.getQos();
        final ByteBuf origMessage = pubMsg.getMessage();
        int count = 0;
        Set<SubscriptionTree.ClientInfo<Tree1Client>> clientInfos = tree1.match(topic);
        for (SubscriptionTree.ClientInfo<Tree1Client> sub : clientInfos) {
            final MqttQoS qos = (publishingQos.value() > sub.getQos()) ? MqttQoS.valueOf(sub.getQos()) : publishingQos;
            ByteBuf message = origMessage.retainedDuplicate();
            Tree1Client client = sub.getClient();

            MqttPublishMessage publishMessage = createPublishMessage(message, qos, topic, client.getClientId());
            if (publishMessage == null) {
                continue;
            }
            if (qos == MqttQoS.AT_MOST_ONCE) {
                //QoS 0
                routeMessage(client.getClientId(), publishMessage);

            } else {
                int result = routeMessage(client.getClientId(), publishMessage);
                switch (result) {
                    // the client is not online or send to the client success
                    case -1:
                    case 1: {

                        if (!client.isCleanSession() && mqttConfig.isCleanSession()) {
                            addToMessageSaveTask(client.getClientId(), publishMessage.variableHeader().messageId()
                                    , false, pubMsg, 1);
                        }
                    }
                    break;
                    case 0: {
                        // the writebuf to the client is full, add to wait
                        addToMessageSaveTask(client.getClientId(), publishMessage.variableHeader().messageId()
                                , client.isCleanSession(), pubMsg, 1);
                    }
                    break;
                }
            }
            count++;
        }
        origMessage.release();
        return count;
    }

    private void addToMessageSaveTask(String clientId, int messageId, boolean cleanSession, StoredMessage message, int round) {
        WaitAckMessage waitAckMessage = new WaitAckMessage();
        waitAckMessage.setCleanSession(cleanSession);
        Qos1Message qos1Message = new Qos1Message();
        qos1Message.setToClientId(clientId);
        qos1Message.setData(message);
        qos1Message.setRound(round);
        qos1Message.setMsgId(messageId);
        waitAckMessage.setMessage(qos1Message);
        subMessageSaveTask.add(waitAckMessage);
    }

    private MqttPublishMessage createPublishMessage(ByteBuf payload, MqttQoS qos, String topic, String clientId) {
        ConnectionDescriptor connDescr = clientIdMap.get(clientId);
        MqttFixedHeader fixedHeader = new MqttFixedHeader(MqttMessageType.PUBLISH, false, qos
                , false, 0);
        MqttPublishVariableHeader varHeader = null;

        if (qos.value() > AT_MOST_ONCE.value()) {
            if (connDescr != null) {
                varHeader = new MqttPublishVariableHeader(topic, connDescr.nextPacketID());
            } else {
                AtomicInteger msgId = msgIdMap.get(clientId);
                if (msgId != null) {
                    varHeader = new MqttPublishVariableHeader(topic, msgId.getAndIncrement());
                } else {
                    logger.info("error occured when send sub message to offline clients");
                    return null;
                }
            }
        } else {
            varHeader = new MqttPublishVariableHeader(topic, 0);
        }
        return new MqttPublishMessage(fixedHeader, varHeader, payload);
    }


    /**
     * 给某个Client 发送一条Publish 消息
     *
     * @param toClientId     接收消息的Client的ClientId
     * @param publishMessage 待下发的消息
     * @return int 发送成功，返回1;设备不在线，返回-1;设备在线，但是缓冲队列已满，返回0
     */
    private int routeMessage(String toClientId, MqttPublishMessage publishMessage) {

        String topic = publishMessage.variableHeader().topicName();
        ConnectionDescriptor connDescr = clientIdMap.get(toClientId);
        if (connDescr == null) {
            logger.info("send message to subscriber failed because unknow error, maybe the client is not on line. topic:{}, clientId:{}", topic, toClientId);
            return -1;
        }

        logger.info("send publish message to {} on topic {}", toClientId, topic);

        Channel channel = connDescr.channel;
        if (channel.isWritable()) {
            String mqttId = NettyUtils.mqttId(channel);
            //if channel is writable don't enqueue
            channel.writeAndFlush(publishMessage);

            // notify that this broker route a message to client
            dataHandler.onRouteMessageToClient(toClientId, mqttId, topic);

            return 1;
        } else {
            logger.info("send message to subscriber failed because writebuf is full. topic:{}, clientId:{}", topic, toClientId);
            return 0;
        }
    }

    public void sendPubAck(String clientId, int messageId) {
        logger.debug("send pub ack to client. clientId = {}, messageId = {}", clientId, messageId);
        MqttFixedHeader mqttFixedHeader = new MqttFixedHeader(MqttMessageType.PUBACK, false, AT_MOST_ONCE,
                false, 0);
        MqttMessageIdVariableHeader variableHeader = MqttMessageIdVariableHeader.from(messageId);
        MqttPubAckMessage pubAckMessage = new MqttPubAckMessage(mqttFixedHeader, variableHeader);
        try {
            ConnectionDescriptor conn = clientIdMap.get(clientId);
            if (conn == null) {
                logger.error("send PubAck to client {} failed, because client is not connect to this broker", clientId);
                return;
            }
            if (conn.channel.isWritable()) {
                conn.channel.writeAndFlush(pubAckMessage);
            } else {
                logger.error("send pub ack failed. because the write buf is full. clientId = {}, messageId = {}", clientId, messageId);
            }

        } catch (Throwable t) {
            logger.error("sendPubAck failed, {}", t.getMessage());
        }
    }

    /**
     * send publish ack message to the client by msgId
     *
     * @param mqttId    mqttId
     * @param messageId msgId
     */
    public void sendPubAckByMqttId(String mqttId, int messageId) {
        String clientId = mqttIdMap.get(mqttId);
        if (clientId != null) {
            sendPubAck(clientId, messageId);
        }
    }

    /**
     * Second phase of a publish QoS2 protocol, sent by publisher to the broker. Search the stored message and publish
     * to all interested subscribers.
     */
    public void processPubRel(Channel channel, MqttMessage msg) {
        logger.info("process PubRel(...) this broker dosn't suppurt QoS2");
        disconnect(channel);
    }

    public void processPubRec(Channel channel, MqttMessage msg) {
        logger.info("process processPubRec(..) this broker dosn't suppurt QoS2");
        disconnect(channel);
    }

    public void processPubComp(Channel channel, MqttMessage msg) {
        logger.info("process processPubComp(...) this broker dosn't suppurt QoS2");
        disconnect(channel);
    }

    private void disconnect(Channel channel) {
        logger.info("disconnect(...) is called, start disconnect channel: {}", channel);
        try {
            if (channel == null) {
                return;
            }
            channel.flush();
            String clientId = NettyUtils.clientId(channel);
            String mqttId = NettyUtils.mqttId(channel);
            Boolean cleanSession = NettyUtils.cleanSession(channel);

            if (clientId == null) {
                return;
            }

            ConnectionDescriptor conn = clientIdMap.get(clientId);
            if (conn == null) {
                return;
            }

            if (conn.channel == null) {
                return;
            }

            if (clientId == null || mqttId == null) {
                return;
            }

            if (conn.channel.equals(channel)) {

                authorizator.cleanProxy(mqttId);
                authenticator.onClientDisconnected(clientId, mqttId);
                executorService.execute(() -> {
                    if (messageHandler != null) {
                        messageHandler.onClientDisconnected(clientId, mqttId, cleanSession);
                    }
                    if (dataHandler != null) {
                        dataHandler.onClientDisconnected(clientId, mqttId, cleanSession);
                    }
                });

                if (mqttConfig.isSubscribe()) {
                    // 说明 订阅树 tree1 的清理工作 在messageHandler中完成
                    if (mqttConfig.isCleanSession() && !cleanSession) {
                        msgIdMap.put(clientId, new AtomicInteger(conn.nextPacketID()));
                    }
                }

                if (mqttId != null) {
                    mqttIdMap.remove(mqttId);
                }
                if (clientId != null) {
                    clientIdMap.remove(clientId);
                    clientCount.decrementAndGet();
                }
                channel.close();
                logger.info("disconnect client:{} finished, channel:{}", clientId, channel);
            }
        } catch (Exception e) {
            logger.error("error occured when disconnect. {}", e.getMessage());
        }
    }

    public void processDisconnect(Channel channel) throws InterruptedException {
        disconnect(channel);
    }

    public void processConnectionLost(String clientId, Channel channel) {
        if (clientId == null || channel == null) {
            return;
        }

        disconnect(channel);
    }

    /**
     * Remove the clientId from topic subscription, if not previously subscribed,
     * doesn't reply any error
     */
    public void processUnsubscribe(Channel channel, MqttUnsubscribeMessage msg) {
        // 如果本mqtt broker未启用订阅，则直接关闭此连接
        if (!mqttConfig.isSubscribe()) {
            disconnect(channel);
            return;
        }
        List<String> topics = msg.payload().topics();
        int messageId = msg.variableHeader().messageId();
        String clientId = NettyUtils.clientId(channel);
        boolean cleanSession = NettyUtils.cleanSession(channel);

        Tree1Client client = new Tree1Client(clientId, cleanSession);

        for (String topic : topics) {
            int qos = tree1.getQos(topic, client);
            tree1.removeSubscribe(client, topic);
            executorService.execute(() -> {
                if (messageHandler != null) {
                    messageHandler.onTopicUnsubscribed(topic, qos, clientId, cleanSession);
                }
                if (dataHandler != null) {
                    dataHandler.onTopicUnsubscribed(topic, qos, clientId, cleanSession);
                }
            });
            logger.info("client {} unsubscribe filter {}", clientId, topic);
        }
        MqttFixedHeader mqttFixedHeader = new MqttFixedHeader(MqttMessageType.UNSUBACK, false, AT_MOST_ONCE,
                false, 0);
        MqttMessageIdVariableHeader variableHeader = MqttMessageIdVariableHeader.from(messageId);
        MqttUnsubAckMessage ackMessage = new MqttUnsubAckMessage(mqttFixedHeader, variableHeader);
        channel.writeAndFlush(ackMessage);
    }

    public void processSubscribe(Channel channel, MqttSubscribeMessage msg) {
        if (!mqttConfig.isSubscribe()) {
            logger.info("this broker dons't enable subscribe, the connect will disconnect");
            disconnect(channel);
            return;
        }

        String clientId = NettyUtils.clientId(channel);
        String mqttId = NettyUtils.mqttId(channel);
        boolean cleanSession = NettyUtils.cleanSession(channel);

        if (clientId == null || mqttId == null) {
            disconnect(channel);
            return;
        }

        int messageId = msg.variableHeader().messageId();

        List<Integer> subResult = new LinkedList<>();
        List<MqttTopicSubscription> subscriptions = msg.payload().topicSubscriptions();
        for (MqttTopicSubscription req : subscriptions) {
            if (!authorizator.canRead(req.topicName(), clientId, mqttId)) {
                logger.info("client {} subscripe filter {} failed because of it hasn't read credentials", clientId, req.topicName());
                subResult.add(MqttQoS.FAILURE.value());
            } else {
                MqttQoS qos = req.qualityOfService();
                String topic = req.topicName();
                // 如果订阅时的QoS为2，则本订阅无效
                if (qos == MqttQoS.EXACTLY_ONCE || topic.length() > 128) {
                    subResult.add(MqttQoS.FAILURE.value());
                    logger.warn("client {} subscribe filter {} failed, may be your qos is 2 or your topic is too long", clientId, topic);
                    continue;
                }
                Tree1Client client = new Tree1Client(clientId, cleanSession);
                int oldQos = tree1.add(topic, client, qos.value());
                if (oldQos >= 0){
                    subResult.add(qos.value());
                    executorService.execute(() -> {
                        if (messageHandler != null) {
                            messageHandler.onTopicSubscribed(clientId, mqttId, topic, qos.value(), cleanSession, oldQos);
                        }
                        if (dataHandler != null) {
                            dataHandler.onTopicSubscribed(clientId, mqttId, topic, qos.value(), cleanSession, oldQos);
                        }
                    });
                    logger.info("client {} subscribe filter successed", clientId, req.topicName());
                }
                else{
                    logger.warn("client {} subscribe filter {} failed, the filter format is invalid", client, topic);
                    subResult.add(MqttQoS.FAILURE.value());
                    continue;
                }
            }
        }
        //ack the client
        MqttFixedHeader mqttFixedHeader = new MqttFixedHeader(MqttMessageType.UNSUBACK, false, AT_MOST_ONCE,
                false, 0);
        MqttMessageIdVariableHeader variableHeader = MqttMessageIdVariableHeader.from(messageId);
        MqttSubAckPayload payload = new MqttSubAckPayload(subResult);
        MqttSubAckMessage ackMessage = new MqttSubAckMessage(mqttFixedHeader, variableHeader, payload);
        channel.writeAndFlush(ackMessage);

    }


    public boolean offlineMqttId(String mqttId) {
        try {
            if (mqttIdMap.containsKey(mqttId)) {
                String clientId = mqttIdMap.get(mqttId);
                if (clientId == null) {
                    mqttIdMap.remove(mqttId);
                    return false;
                }
                ConnectionDescriptor conn = clientIdMap.get(clientId);
                if (conn != null) {
                    disconnect(conn.channel);
                    return true;
                }
            }
        } catch (Exception e) {
        }
        return false;

    }

    /**
     * offline the client by clientId
     *
     * @param clientId the clientId of the client
     * @return
     */
    public boolean offlineClientId(String clientId) {
        logger.info("offline ClientId is called. clientId = {}", clientId);
        try {
            if (clientId != null) {
                ConnectionDescriptor conn = clientIdMap.get(clientId);
                if (conn != null && conn.channel != null) {
                    disconnect(conn.channel);
                    return true;
                }
            }
        } catch (Exception e) {

        }
        return false;
    }

    private static final class ConnectionDescriptor {
        private final Channel channel;
        private AtomicInteger maxMsgId = new AtomicInteger(new Random(0xffff).nextInt(65000));

        private ConnectionDescriptor(Channel session) {
            this.channel = session;
        }

        private int nextPacketID() {
            int nextPacketId = maxMsgId.get();
            if (nextPacketId > 65000) {
                while (true) {
                    nextPacketId = maxMsgId.get();
                    if (nextPacketId > 65000) {
                        if (maxMsgId.compareAndSet(nextPacketId, nextPacketId - 6500)) {
                            break;
                        }
                    } else {
                        break;
                    }
                }
            }
            return maxMsgId.incrementAndGet();
        }
    }
}
